package com.demo.kafka.constants;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * Kafka 延时级别
 * 级别分别是1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
 * @author cyj
 */
@Slf4j
public enum DelayTopicEnum {
    /**
     * 1s
     */
    TOPIC_1(KafkaConstants.getDelayTopic(1),1000),
    /**
     * 5s
     */
    TOPIC_2(KafkaConstants.getDelayTopic(2),5000),
    /**
     * 10s
     */
    TOPIC_3(KafkaConstants.getDelayTopic(3),10000),
    /**
     * 30s
     */
    TOPIC_4(KafkaConstants.getDelayTopic(4),30000),
    /**
     * 1m
     */
    TOPIC_5(KafkaConstants.getDelayTopic(5),60000),
    /**
     * 2m
     */
    TOPIC_6(KafkaConstants.getDelayTopic(6),120000),
    /**
     * 3m
     */
    TOPIC_7(KafkaConstants.getDelayTopic(7),180000),
    /**
     * 4m
     */
    TOPIC_8(KafkaConstants.getDelayTopic(8),240000),
    /**
     * 5m
     */
    TOPIC_9(KafkaConstants.getDelayTopic(9),300000),
    /**
     * 6m
     */
    TOPIC_10(KafkaConstants.getDelayTopic(10),360000),
    /**
     * 7m
     */
    TOPIC_11(KafkaConstants.getDelayTopic(11),420000),
    /**
     * 8m
     */
    TOPIC_12(KafkaConstants.getDelayTopic(12),480000),
    /**
     * 9m
     */
    TOPIC_13(KafkaConstants.getDelayTopic(13),540000),
    /**
     * 10m
     */
    TOPIC_14(KafkaConstants.getDelayTopic(14),600000),
    /**
     * 20m
     */
    TOPIC_15(KafkaConstants.getDelayTopic(15),1200000),
    /**
     * 30m
     */
    TOPIC_16(KafkaConstants.getDelayTopic(16),1800000),
    /**
     * 1h
     */
    TOPIC_17(KafkaConstants.getDelayTopic(17),3600000),
    /**
     * 2h
     */
    TOPIC_18(KafkaConstants.getDelayTopic(18),7200000);

    private final String topic;
    private final Long time;

    DelayTopicEnum(String topic, long time) {
        this.topic = topic;
        this.time = time;
    }

    public String getTopic() {
        return topic;
    }

    public Long getTime() {
        return time;
    }

    /**
     * 计算时间对应延时级别
     * @param millis 毫秒
     * @return 延时级别枚举对象
     */
    public static String calculateDefault(long millis) {
        String topic = "";
        // 通过循环计算找出延时等级，算出第一个延时等级终止循环
        for (int i = DelayTopicEnum.values().length - 1; i >= 0 ; i--) {
            Long delayTime = DelayTopicEnum.values()[i].time;
            long l = millis / delayTime;
            log.debug("时间：{}", l);
            if (l > 0) {
                topic = DelayTopicEnum.values()[i].getTopic();
                break;
            }
        }

        return topic;
    }

    /**
     * 根据分区获取对应的延时时间
     * @param topic 分区
     * @return 延时时间
     */
    public static long getDelayTime(String topic) {
        for (DelayTopicEnum value : DelayTopicEnum.values()) {
            if (value.topic.equalsIgnoreCase(topic)) {
                return value.time;
            }
        }
        return 0;
    }

    public static List<String> getTopics() {
        return Stream.of(DelayTopicEnum.values()).map(DelayTopicEnum::getTopic).collect(Collectors.toList());
    }

    public static DelayTopicEnum getDelayTopicEnum(String topic) {
        return Stream.of(DelayTopicEnum.values()).filter(s -> StringUtils.equalsIgnoreCase(s.getTopic(), topic)).findFirst().get();
    }
}
